﻿using SuperX.Common.Helper;
using SuperX.Common.Log;
using SuperX.Helper;
using SuperX.SuperSocket.Socket.Client;
using SuperX.SuperSocket.Socket.Message;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace SuperX.SuperSocket.Socket.Rpc
{
    public class RpcClient
    {
        private SocketClient socket;

        private bool isDisposed = false;

        /// <summary>
        /// 接收命令缓冲区
        /// </summary>
        private readonly ConcurrentDictionary<string, XSocketMessage> ResponseBuffer
            = new ConcurrentDictionary<string, XSocketMessage>();

        private readonly ConcurrentDictionary<string, AutoResetEvent> CommandWaiter
            = new ConcurrentDictionary<string, AutoResetEvent>();

        private object _objLock = new object();

        private ConcurrentDictionary<string, Response<object>> LastResponseLog =
                        new ConcurrentDictionary<string, Response<object>>();

        private ConcurrentQueue<RpcRequest> requestBuff = new ConcurrentQueue<RpcRequest>();
        private SocketClientConfig config { get; set; }

        public bool IsConnect => socket != null && socket.IsConnected;

        public RpcClient(string hostName, int port, bool isEnableHeartBeat = false)
            : this(new SocketClientConfig() { Ip = hostName, Port = port, IsEnableCheckHeartbeat = isEnableHeartBeat })
        {
        }

        public RpcClient(SocketClientConfig clientConfig)
        {
            config = clientConfig;
            socket = new SocketClient(clientConfig);
            socket.MessageRecevied += Socket_MessageRecevied;
            socket.Init();
            new Thread(SendRpcRequest).Start();
            var result = socket.Connect();
        }

        public bool IsConnected()
        {
            return socket.IsConnected;
        }

        /// <summary>
        /// 检查等待发送请求任务的线程
        /// </summary>
        private void SendRpcRequest()
        {
            while (!isDisposed)
            {
                try
                {
                    while (requestBuff.TryDequeue(out var req))
                    {
                        socket.Send(req);
                    }
                    Thread.Sleep(10);
                }
                catch (Exception)
                {
                }
            }
        }

        /// <summary>
        /// 重连
        /// </summary>
        /// <returns></returns>
        public bool Reconnect()
        {
            if (!socket.IsConnected)
            {
                try
                {
                    socket.DisConnect();
                    socket.Connect();
                }
                catch { }
            }
            return socket.IsConnected;
        }

        /// <summary>
        /// 接收信息事件
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void Socket_MessageRecevied(object sender, MessageArgs e)
        {
            if (e?.AppMessage?.Content?.StartsWith(XSocketMessage.ConnectedConst) == true ||
                e?.AppMessage?.Content?.StartsWith(XSocketMessage.ErrorConst) == true ||
                e?.AppMessage?.Content?.StartsWith(XSocketMessage.DisconnectedConst) == true)
            {
                return;
            }

            var response = e?.AppMessage?.GetObject<Response<object>>();
            lock (_objLock)
            {
                //1.判断是否是之前的请求返回
                if (response?.RequestId != null)
                {
                    if (CommandWaiter?.ContainsKey(response.RequestId) == true)
                    {
                        ResponseBuffer?.TryAdd(response.RequestId, (XSocketMessage)e?.AppMessage);
                        CommandWaiter.TryRemove(response.RequestId, out AutoResetEvent waiter);
                        waiter.Set();   //释放等待
                    }
                }
                //2.判断是否是要调用server这边的方法，调用应答
                //InvokeRequest(e);
            }
        }

        #region Rpc请求

        /// <summary>
        /// 异步调用
        /// </summary>
        public Task<T> CallAsync<T>(RpcPara rpcPara, AsyncCallback callback, int timeout = 5000)
        {
            var task = new Task<T>(
                () => Call<T>(rpcPara, timeout)
                );
            task.Start();
            return task;
        }

        /// <summary>
        /// 同步调用返回
        /// 调用流程
        /// 1.RpcPara增加Guid头部信息；2.rpc信息放到请求队列里；3.队列轮询取出执行；4.等待执行结果；5.通过Guid头部信息找回结果；6.Json反序列化实体
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <returns></returns>
        public T Call<T>(RpcPara rpcPara, int timeout = 5000)
        {
            try
            {
                var request = new RpcRequest() { RpcPara = rpcPara };
                var waiter = new AutoResetEvent(false);
                CommandWaiter.TryAdd(request.RequestId, waiter);

                if (rpcPara.Headers == null)
                {
                    rpcPara.Headers = new Dictionary<string, string>();
                }
                rpcPara.Headers.Add("RequestId", request.RequestId);
                //socket.Send(request);
                requestBuff.Enqueue(request);
                if (waiter.WaitOne(timeout))    //异步调用等待调用结果
                {
                    if (ResponseBuffer.TryRemove(request.RequestId, out XSocketMessage reponseMsg))
                    {
                        var response = reponseMsg.GetObject<Response<T>>();
                        Response<object> responseObj;
                        if (response != null && response.Data != null)
                        {
                            responseObj = new Response<object>() { Data = response.Data };
                        }
                        else
                        {
                            responseObj = reponseMsg.GetObject<Response<object>>();
                        }
                        LastResponseLog.AddOrUpdate(rpcPara.MethodName, responseObj,
                            (key, oldValue) => responseObj);
                        if (responseObj.IsSuccess != true)
                        {
                            ShowError(response?.Error);
                            return default;
                        }
                        else
                        {
                            if (response == null)
                            {
                                return default;
                            }
                            else
                            {
                                return response.Data;
                                //                    return response.Data;
                            }
                        }
                    }
                }
                else
                {
                    //Reconnect();
                    LastResponseLog[rpcPara.MethodName] = null;
                    lock (_objLock)
                    {
                        CommandWaiter.TryRemove(request.RequestId, out waiter);
                    }

                    var responseObj = new Response<object>()
                    {
                        IsSuccess = false,
                        Error = new Error("rpc request timeout", 10)
                    };
                    Logger.Trace($"rpc request timeout method info:{JsonHelper.JsonSerialize(rpcPara)}");

                    LastResponseLog.AddOrUpdate(rpcPara.MethodName, responseObj,
                        (key, oldValue) => responseObj);
                }
            }
            catch
            {
                //Reconnect();
            }
            return default;
        }

        #endregion Rpc请求

        #region Rpc应答

        private Dictionary<string, MethodInvokeInfo> _commandMethods = new Dictionary<string, MethodInvokeInfo>();

        public void RegistCommandMethods(Dictionary<string, MethodInvokeInfo> methods)
        {
            foreach (var item in methods)
            {
                if (_commandMethods.ContainsKey(item.Key))
                {
                    Logger.Warn($"在RpcClient中，已注册有该方法：{item.Key}");
                    continue;
                }
                _commandMethods.Add(item.Key, item.Value);
            }
        }

        private void InvokeRequest(MessageArgs e)
        {
            Task.Factory.StartNew(() =>
            {
                var request = e.AppMessage.GetObject<RpcRequest>();
                if (request?.RequestId != null)
                {
                    var methodName = request.RpcPara.MethodName;
                    MethodInvokeInfo method = null;
                    if (_commandMethods.ContainsKey(request.RpcPara.MethodName))
                    {
                        method = _commandMethods[request.RpcPara.MethodName];
                    }
                    Response<object> response = null;

                    if (method != null)
                    {
                        var result = method.Invoke(request.RpcPara.Paras);//调用函数

                        if (result is Response<object> tempresponse)
                        {
                            response = new Response<object>(tempresponse) { RequestId = request.RequestId };
                        }
                        else
                        {
                            response = new Response<object>() { RequestId = request.RequestId, Data = result };
                        }
                    }
                    else
                    {
                        response = new Response<object>()
                        {
                            IsSuccess = false,
                            RequestId = request.RequestId,
                            Error = new Error($"Gateway Method not found:{methodName}", 0)
                        };
                    }
                    e.AppSession.Send(JsonHelper.JsonSerialize(response));
                }
            });
        }

        #endregion Rpc应答

        public void Dispose()
        {
            isDisposed = true;
            socket.MessageRecevied -= Socket_MessageRecevied;
            socket?.DisConnect();
        }

        private void ShowError(Error error)
        {
            if (error != null)
            {
                if (error.InnerError != null)
                {
                    ShowError(error.InnerError);
                }
                else
                {
                    Logger.Warn(error.Message);
                }
            }
        }

        public Response<object> GetLastRequestResult(string methodName)
        {
            if (LastResponseLog.ContainsKey(methodName))
            {
                return LastResponseLog[methodName];
            }
            return null;
        }
    }
}